package com.bigdata.warehouse.dws;

import com.bigdata.warehouse.entity.OwnerBehavior;
import com.bigdata.warehouse.sink.DwsOwnerBehaviorClickhouseSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.sql.Timestamp;

public class DwsOwnerBehavior {
    public static void main(String[] args) throws Exception {
        //1.获取Stream的执行环境
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        senv.setParallelism(1);

        //2.创建表的执行环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(senv);

        //开启checkpoint容错
//        senv.enableCheckpointing(60000);
//        senv.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        senv.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
//        senv.getCheckpointConfig().setCheckpointTimeout(10000);
//        senv.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

        //设置状态后端
        //（1）开启RocksDB
//        senv.setStateBackend(new EmbeddedRocksDBStateBackend());
//        //（2）设置checkpoint存储位置
//        senv.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("hdfs://mycluster/flink/checkpoints"));

        //3.读取门禁明细表
        tEnv.executeSql("CREATE TABLE dwd_entrance_guard_log (\n" +
                "id STRING,\n" +
                "ownerId INT,\n" +
                "openTime STRING,\n" +
                "cId BIGINT,\n" +
                "openMode STRING,\n" +
                " time_ltz AS TO_TIMESTAMP_LTZ(UNIX_TIMESTAMP(openTime),0),\n" +
                " WATERMARK FOR time_ltz AS time_ltz - INTERVAL '1' SECOND"+
                ") WITH (\n" +
                "'connector' = 'kafka',\n" +
                "'topic' = 'dwd_entrance_guard_log',\n" +
                "'properties.bootstrap.servers' = 'master:9092,slave1:9092,slave2:9092',\n" +
                " 'properties.group.id' = 'dwd_entrance_guard_log',\n" +
                " 'scan.startup.mode' = 'earliest-offset',\n" +
                " 'format' = 'json'\n" +
                ")\n");
        //4.读取用户维表
        tEnv.executeSql("CREATE TABLE dim_base_ower(\n" +
                " id INT,\n" +
                " roomId INT,\n" +
                " owerCode STRING,\n" +
                " owerName STRING,\n" +
                " owerSex STRING,\n" +
                " owerAge INT,\n" +
                " owerIdcard STRING,\n" +
                " owerTel STRING,\n" +
                " owerHomeplace STRING,\n" +
                " owerWorkplace STRING,\n" +
                " remark STRING,\n" +
                " createTime TIMESTAMP,\n" +
                " operatorTime TIMESTAMP,\n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ")  WITH (\n" +
                " 'connector' = 'jdbc',\n" +
                " 'url' = 'jdbc:mysql://master:3306/sca?useUnicode=true&characterEncoding=utf8',\n" +
                " 'table-name' = 'dim_base_ower',\n" +
                " 'username' = 'hive',\n" +
                " 'password' = 'hive'\n" +
                ")\n");
        //5.读取房间维度表
        tEnv.executeSql("CREATE TABLE `dim_base_room`(\n" +
                " `id` INT,\n" +
                " `houseId` INT,\n" +
                " `roomCode` STRING,\n" +
                " `roomName` STRING,\n" +
                " `peopleNum` INT,\n" +
                " `remark` STRING,\n" +
                " PRIMARY KEY (`id`) NOT ENFORCED\n" +
                ") WITH (\n" +
                " 'connector' = 'jdbc',\n" +
                " 'url' = 'jdbc:mysql://master:3306/sca?useUnicode=true&characterEncoding=utf8',\n" +
                " 'username' = 'hive',\n" +
                " 'password' = 'hive',\n" +
                " 'table-name' = 'dim_base_room'\n" +
                ")\n");
        //6.读取楼栋维度表
        tEnv.executeSql("CREATE TABLE `dim_base_house`(\n" +
                " `id` INT,\n" +
                " `cid` INT,\n" +
                " `houseCode` STRING,\n" +
                " `houseName` STRING,\n" +
                " `buildStartTime` TIMESTAMP,\n" +
                " `buildEndTime` TIMESTAMP,\n" +
                " `houseArea` FLOAT,\n" +
                " `remark` STRING,\n" +
                " PRIMARY KEY (`id`) NOT ENFORCED\n" +
                ")  WITH (\n" +
                " 'connector' = 'jdbc',\n" +
                " 'url' = 'jdbc:mysql://master:3306/sca?useUnicode=true&characterEncoding=utf8',\n" +
                " 'username' = 'hive',\n" +
                " 'password' = 'hive',\n" +
                " 'table-name' = 'dim_base_house'\n" +
                ")\n");
        //7.读取社区维度表
        tEnv.executeSql("CREATE TABLE dim_base_community(\n" +
                " id INT,\n" +
                " communityName STRING,\n" +
                " district_id INT,\n" +
                " city_id INT,\n" +
                " communityLocation STRING,\n" +
                " remark STRING,\n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                " 'connector' = 'jdbc',\n" +
                " 'url' = 'jdbc:mysql://master:3306/sca?useUnicode=true&characterEncoding=utf8',\n" +
                " 'table-name' = 'dim_base_community',\n" +
                " 'username' = 'hive',\n" +
                " 'password' = 'hive'\n" +
                ")\n");
        //8.读取地区维度表
        tEnv.executeSql("CREATE TABLE `dim_base_district` (\n" +
                " `id` INT,\n" +
                " `pid` INT,\n" +
                " `code` STRING,\n" +
                " `name` STRING,\n" +
                " `remark` STRING,\n" +
                " `create_time` TIMESTAMP,\n" +
                " `update_time` TIMESTAMP,\n" +
                " `status` SMALLINT,\n" +
                " `level` SMALLINT,\n" +
                " PRIMARY KEY (`id`)  NOT ENFORCED\n" +
                ") WITH (\n" +
                " 'connector' = 'jdbc',\n" +
                " 'url' = 'jdbc:mysql://master:3306/sca?useUnicode=true&characterEncoding=utf8',\n" +
                " 'table-name' = 'dim_base_district',\n" +
                " 'username' = 'hive',\n" +
                " 'password' = 'hive' \n" +
                ")\n");
        //9.窗口计算
        Table windowTable = tEnv.sqlQuery(
                "select  "+
                        "TUMBLE_START(time_ltz,INTERVAL '1' second) as stt,"+
                        "TUMBLE_END(time_ltz,INTERVAL '1' second) as edt,"+
                        "CURRENT_TIMESTAMP as ts,"+
                        "cId,"+
                        "ownerId,"+
                        "sum(if(openMode='1',1,0)) entrance_card_cnt,"+
                        "sum(if(openMode='2',1,0)) entrance_pw_cnt,"+
                        "sum(if(openMode='3',1,0)) entrance_face_cnt,"+
                        "sum(if(openMode='4',1,0)) entrance_remote_cnt "+
                        " from dwd_entrance_guard_log GROUP BY cId,ownerId, TUMBLE(time_ltz,INTERVAL '1' second) "
        );
        //windowTable.execute().print();

        tEnv.createTemporaryView("windowTable",windowTable);
        //10.维表关联
        Table joinTable = tEnv.sqlQuery(
                "select "+
                        "egl.ownerId,"+
                        "ol.owerSex,"+
                        "ol.owerAge,"+
                        "ol.owerHomeplace,"+
                        "ol.owerWorkplace,"+
                        "c.id as community_id,"+
                        "c.communityName,"+
                        "c.city_id,"+
                        "c.district_id,"+
                        "d.name as  district_name,"+
                        "egl.entrance_card_cnt,"+
                        "egl.entrance_pw_cnt,"+
                        "egl.entrance_face_cnt,"+
                        "egl.entrance_remote_cnt,"+
                        "egl.stt,"+
                        "egl.edt,"+
                        "egl.ts "+
                        " from windowTable egl "+
                        "join dim_base_ower ol on egl.ownerId=ol.id "+
                        "join dim_base_room r on ol.roomId=r.id "+
                        "join dim_base_house h on r.houseId=h.id "+
                        "join dim_base_community c on h.cid=c.id "+
                        "join dim_base_district d on c.district_id=d.id"
        );
        //joinTable.execute().print();


        //11.查询结果写入clickhouse
        DataStream<OwnerBehavior> obDataStream =tEnv.toDataStream(joinTable,OwnerBehavior.class);
        String sql = "insert into default.dws_owner_behavior (ownerId,owerSex,owerAge,owerHomeplace," +
                "owerWorkplace,community_id,communityName,city_id,district_id,district_name," +
                "entrance_card_cnt,entrance_pw_cnt,entrance_face_cnt,entrance_remote_cnt," +
                "stt,edt,ts) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
        DwsOwnerBehaviorClickhouseSink jdbcSink = new DwsOwnerBehaviorClickhouseSink(sql);
        obDataStream.addSink(jdbcSink);
        System.out.println("-----------------------------------------0");
        senv.execute("DwsOwnerBehavior");
    }
}
